package org.lock;

import jdk.nashorn.internal.codegen.CompilerConstants;
import org.apache.curator.RetryPolicy;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.framework.recipes.locks.InterProcessLock;
import org.apache.curator.framework.recipes.locks.InterProcessMutex;
import org.apache.curator.retry.ExponentialBackoffRetry;
import org.junit.Test;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.*;
import java.util.function.Consumer;


public class CuratorLock {



    @Test
    public void testlock() throws  Exception{

        //重试策略，初始时间为6s，重试次数为2次
        RetryPolicy retryPolicy = new ExponentialBackoffRetry(6000, 2);

        CuratorFramework cf = CuratorFrameworkFactory.builder()
                .connectString("10.10.10.92:2181").sessionTimeoutMs(9000)
                .retryPolicy(retryPolicy).build();

        cf.start();

        // 分布式锁
        InterProcessLock lock=new InterProcessMutex(cf,"/zkLock") ;

        //创建缓存线程池
        ExecutorService executorService= Executors.newCachedThreadPool();

         Consumer<InterProcessMutex> consumer = new Consumer<InterProcessMutex>() {
            public void accept(InterProcessMutex interProcessMutex) {
                List<Callable<String>> list =new ArrayList<>();
                Callable<String> callable=new Callable<String>() {
                    @Override
                    public String call() throws Exception {
                        try {


                        interProcessMutex.acquire();
                        System.err.println("开始业务处理");
                        System.err.println("获取锁：："+Thread.currentThread().getName());

                        System.err.println("业务处理结束");

                        }catch (Exception e){
                               e.printStackTrace();
                        }finally {
                            //释放锁
                            interProcessMutex.release();
                            System.err.println("释放锁：："+Thread.currentThread().getName());
                        }
                        return "true";
                    }
                };
                for (int i = 0; i <10 ; i++) {
                 list.add(callable);
                    try {
                        Thread.sleep(2000);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                }

                try {
                    executorService.invokeAll(list);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        };




        consumer.accept((InterProcessMutex) lock);
        executorService.shutdown();
        cf.close();
    }
}
